package com.lzwk.app.dws.statLog;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lzwk.beans.StatLogNZ;
import com.lzwk.config.JobConfig;
import com.lzwk.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * @Author: CC
 * @Date: 2022/3/9 11:28
 */
public class neZha {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        JobConfig prod = new JobConfig("dev", env);

        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(prod.statLogNzTopic, prod.statLogNzGroupId, prod.statLogKafkaServers));
        //处理迟到数据
        OutputTag<String> outputTag = new OutputTag<String>("output") {
        };

        SingleOutputStreamOperator<StatLogNZ> StatLogNZDS = kafkaDS.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            String action = jsonObject.getString("action");
            Long bookingId = jsonObject.getLong("booking_id");
            Long linkId = jsonObject.getLong("link_id");
            Long accountId = jsonObject.getLong("account_id");
            Long ts = jsonObject.getLong("ts");
            return new StatLogNZ(action, bookingId, linkId, accountId, ts);
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<StatLogNZ>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                .withTimestampAssigner(new SerializableTimestampAssigner<StatLogNZ>() {
                    @Override
                    public long extractTimestamp(StatLogNZ element, long l) {
                        return element.getTs();
                    }
                }).withIdleness(Duration.ofSeconds(15))
        );

        //开窗
        StatLogNZDS.keyBy(StatLogNZ::getAccount_id).print();

        env.execute("test");
    }

}
